package org.example.flink;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author: tang
 * @Description: DataStream socket API
 * @Date 2025/2/14 10:25
 */
public class FlinkWordCountStreamSocketMainParallelism {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment
                .createLocalEnvironmentWithWebUI(new Configuration());

        environment.setParallelism(3);

        DataStreamSource<String> streamSource = environment.socketTextStream("127.0.0.1", 7777);

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = streamSource.flatMap(
                (String key, Collector<Tuple2<String, Integer>> collector) -> {
                    String[] worlds = key.split(" ");
                    for (String world : worlds) {
                        collector.collect(Tuple2.of(world, 1));
                    }
                }).returns(Types.TUPLE(Types.STRING, Types.INT))
                // 设置并行度
                .setParallelism(2)
                .keyBy(value -> value.f0);
        keyedStream.sum(1)
                .print();
        environment.execute();
    }

}
